package cn.pink.common.support.tools.rpc;

import cn.hutool.core.lang.Pair;
import cn.hutool.core.util.RandomUtil;
import cn.pink.core.CallPoint;
import cn.pink.core.support.Param;
import cn.pink.core.support.serializer.SerializerUtil;
import cn.pink.dbsrv.RPCProxy;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.net.SocketAddress;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @Author: pink
 * @Date: 2022/7/13 17:51
 */
public class RpcTool {
    private static RpcTool tool = new RpcTool();

    private RpcTool() {

    }

    public static RpcTool getInstance() {
        return tool;
    }

    private Vertx vertx;

    private List<SocketAddress> address = new ArrayList<>();

    public void init(Vertx vertx) {
        this.vertx = vertx;
    }

    /**
     * 随机一个rpc服务地址
     */
    private SocketAddress getRpcAddress() {
        return RandomUtil.randomEle(address);
    }

    //---------------------------------- rpc method ----------------------------------

    /**
     * 查找human
     */
    public Future<String> findHuman(String id) {
        Promise<String> promise = Promise.promise();

        RPCProxy.DBServiceProxy proxy = RPCProxy.DBServiceProxy.newInstance();
        proxy.findHuman(id);

        proxy.listenResult((Param result, Param context) -> {
            promise.complete(result.get());
        });

        return promise.future();
    }

    /**
     * 提交实体数据变化
     */
    public Future<Void> update(String humanId, Map<String, String> update, Set<String> unsets, CallPoint dbPoint) {
        Promise<Void> promise = Promise.promise();

        if((update == null || update.isEmpty()) && (unsets == null || unsets.isEmpty())) {
            return promise.future();
        }

        RPCProxy.DBServiceProxy proxy = RPCProxy.DBServiceProxy.newInstance(dbPoint.nodeId, dbPoint.portId);
        proxy.update(humanId, update, unsets);

        return promise.future();
    }

    public Future<Pair<Long, Long>> idAllot(int max) {
        Promise<Pair<Long, Long>> promise = Promise.promise();

        RPCProxy.DBServiceProxy proxy = RPCProxy.DBServiceProxy.newInstance();
        proxy.idAllot(max);

        proxy.listenResult((Param result, Param context) -> {
            promise.complete(Pair.of(result.get("idMin"), result.get("idMax")));
        });

        return promise.future();
    }

    public Future<Void> cachePersist(String region, String key, byte[] data) {
        Promise<Void> promise = Promise.promise();

        RPCProxy.DBServiceProxy proxy = RPCProxy.DBServiceProxy.newInstance();
        proxy.cachePersist(region, key, data);

        return promise.future();
    }

    public Future<Map<String, Object>> cacheLoad(String region) {
        Promise<Map<String, Object>> promise = Promise.promise();

        RPCProxy.DBServiceProxy proxy = RPCProxy.DBServiceProxy.newInstance();
        proxy.cacheLoad(region);

        proxy.listenResult((Param result, Param context) -> {
            Map<String, byte[]> m = result.get();

            Map<String, Object> map = m.entrySet().stream().collect(Collectors.toMap(
                    Map.Entry::getKey,
                    kv -> SerializerUtil.deserialize(kv.getValue())
            ));

            promise.complete(map);
        });

        return promise.future();
    }
}
